package com.tdy.cdc.app.dwd;

import com.tdy.cdc.app.BaseSqlApp;
import com.tdy.cdc.common.Constant;
import com.tdy.cdc.util.FlinkSinkUtil;
import com.tdy.cdc.util.FlinkSourceUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author NanHuang
 * @Date 2023/3/14
 */
public class App09_DwdMhsSurgicalNursingRecord extends BaseSqlApp {
    public static void main(String[] args) {
        new App09_DwdMhsSurgicalNursingRecord().init(
                5009,
                "App9_DwdMhsSurgicalNursingRecord",
                2
        );
    }

    @Override
    protected void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1 创建源表
        createSourceTables(tableEnv);
        // 2 join
        joinTables(tableEnv);
        // 3 写入Kafka
        writeToKafka(tableEnv);
    }

    private void writeToKafka(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table target (" +
                "OPER_NURSING_BILL_ID              String,"+
                "HOSPITAL_NAME                     String,"+
                "HOSPITAL_CODE                     String,"+
                "AGE_UNIT                          String,"+
                "AGE                               String,"+
                "SEX_NAME                          String,"+
                "SEX_CODE                          String,"+
                "ID_CARD_NUMBER                    String,"+
                "PATIENT_VISIT_CATEGORY_CODE       String,"+
                "PATIENT_VISIT_CATEGORY            String,"+
                "INPATIENT_VISIT_FLOW_ID           String,"+
                "INPATIENT_NO                      String,"+
                "OUTPATIENT_VISIT_FLOW_ID          String,"+
                "OUTPATIENT_NO                     String,"+
                "PATIENT_ID                        String,"+
                "PATIENT_NAME                      String,"+
                "DOMAIN_ID                         String,"+
                "BED_NAME                          String,"+
                "ROOM_NAME                         String,"+
                "VISIT_TIMES                       String,"+
                "INSTITUTIONAL_INCODE              String,"+
                "INSTITUTIONAL_IN                  String,"+
                "AUTHOR_CODE                       String,"+
                "AUTHOR_NAME                       String,"+
                "CUSTODIAN_CODE                    String,"+
                "CUSTODIAN_NAME                    String,"+
                "PATROL_SIGNDATE                   String,"+
                "VISITATORIAL_NURSE_ID             String,"+
                "VISITATORIAL_NURSE_SIGN_NAME      String,"+
                "INSTRUMENT_NURSE_SIGN_DATETIME    String,"+
                "INSTRUMENT_NURSE_ID               String,"+
                "INSTRUMENT_NURSE_SIGN_NAME        String,"+
                "ASSOCIATE_CHECK_ITEM              String,"+
                "ASSOCIATE_NURSE_ID                String,"+
                "ASSOCIATE_NURSE_SIGN_NAME         String,"+
                "ASSOCIATE_NURSE_SIGN_DATETIME     String,"+
                "TRANSPORT_STAFF_SIGN_DATETIME     String,"+
                "TRANSPORT_STAFF_ID                String,"+
                "TRANSPORT_STAFF_SIGN_NAME         String,"+
                "VISIT_DATE                        String,"+
                "BED_NO                            String,"+
                "ROOM_NO                           String,"+
                "DEPT_ID                           String,"+
                "DEPT_NAME                         String,"+
                "WARD_ID                           String,"+
                "WARD_NAME                         String,"+
                "PATIENT_WEIGHT                    String,"+
                "ABO_BLOOD_CODE                    String,"+
                "ABO_BLOOD_NAME                    String,"+
                "RH_BLOOD_CODE                     String,"+
                "RH_BLOOD_NAME                     String,"+
                "NURSE_GRADE_CODE                  String,"+
                "NURSE_TYPE_CODE                   String,"+
                "NURSE_TYPE_NAME                   String,"+
                "PATROL_DATE_AO                    String,"+
                "PATROL_CODE_AO                    String,"+
                "PATROL_NAME_AO                    String,"+
                "SCRUB_DATE_AO                     String,"+
                "SCRUB_CODE_AO                     String,"+
                "SCRUB_NAME_AO                     String,"+
                "OPER_GOOD_AO                      String,"+
                "PRE_OPER_STOCKTAKE_FLAG           String,"+
                "PATROL_DATE_AC                    String,"+
                "PATROL_CODE_AC                    String,"+
                "SCRUB_CODE_AC                     String,"+
                "SCRUB_NAME_AC                     String,"+
                "IN_OPER_GOODS_NAME                String,"+
                "BEFORE_CHECK_MARK                 String,"+
                "PATROL_DATE_BC                    String,"+
                "PATROL_CODE_BC                    String,"+
                "PATROL_NAME_BC                    String,"+
                "SCRUB_CODE_BC                     String,"+
                "SCRUB_NAME_BC                     String,"+
                "OPER_GOOD_BC                      String,"+
                "AFTER_CHECK_MARK                  String,"+
                "OPER_START_DATETIME               String,"+
                "OPER_END_DATETIME                 String,"+
                "OPER_DOCTOR_ID                    String,"+
                "OPER_DOCTOR_NAME                  String,"+
                "OPER_SITE_NAME                    String,"+
                "OPER_SITE_CODE                    String,"+
                "OPER_CODE                         String,"+
                "OPER_NAME                         String,"+
                "ADMIT_ROOM_DATE                   String,"+
                "OUT_OPER_DATE                     String,"+
                "PREPARATIONS                      String," +
                "VISIT_ID                          String,"+
                "primary key (OPER_NURSING_BILL_ID) not enforced" +
                ")" +
                FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_SURGICAL_NURSING_RECORD));
        tableEnv.executeSql("insert into target select * from join_result");

    }

    private void joinTables(StreamTableEnvironment tableEnv) {
        //1 设置数据的TTL（暂时10min）
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60));
        //2 join操作
        Table joinResult = tableEnv.sqlQuery("select " +
                "s.unique_id                                                      AS OPER_NURSING_BILL_ID,"+
                "hospital_name                                                    AS HOSPITAL_NAME,"+
                "hospital_code                                                    AS HOSPITAL_CODE,"+
                "age_unit                                                         AS AGE_UNIT,"+
                "age                                                              AS AGE,"+
                "sex                                                              AS SEX_NAME,"+
                "sex_code                                                         AS SEX_CODE,"+
                "identify_no                                                      AS ID_CARD_NUMBER,"+
                "patient_typecode                                                 AS PATIENT_VISIT_CATEGORY_CODE,"+
                "patient_type                                                     AS PATIENT_VISIT_CATEGORY,"+
                "case when patient_typecode ='3' then visit_id else '' end       AS INPATIENT_VISIT_FLOW_ID,"+
                "inpatient                                                        AS INPATIENT_NO,"+
                "case when patient_typecode <> '3' then visit_id else '' end     AS OUTPATIENT_VISIT_FLOW_ID,"+
                "out_patient_id                                                   AS OUTPATIENT_NO,"+
                "patient_id                                                       AS PATIENT_ID,"+
                "patient_name                                                     AS PATIENT_NAME,"+
                "s.domain_id                                                      AS DOMAIN_ID,"+
                "bed_name                                                         AS BED_NAME,"+
                "ward_name                                                        AS ROOM_NAME,"+
                "visit_times                                                      AS VISIT_TIMES,"+
                "institutional_incode                                             AS INSTITUTIONAL_INCODE,"+
                "institutional_in                                                 AS INSTITUTIONAL_IN,"+
                "author_code                                                      AS AUTHOR_CODE,"+
                "author_name                                                      AS AUTHOR_NAME,"+
                "custodian_code                                                   AS CUSTODIAN_CODE,"+
                "custodian_name                                                   AS CUSTODIAN_NAME,"+
                "patrol_signdate                                                  AS PATROL_SIGNDATE,"+
                "patrol_code                                                      AS VISITATORIAL_NURSE_ID,"+
                "patrol_name                                                      AS VISITATORIAL_NURSE_SIGN_NAME,"+
                "instrument_signdate                                              AS INSTRUMENT_NURSE_SIGN_DATETIME,"+
                "instrument_code                                                  AS INSTRUMENT_NURSE_ID,"+
                "instrument_name                                                  AS INSTRUMENT_NURSE_SIGN_NAME,"+
                "handover_checkup                                                 AS ASSOCIATE_CHECK_ITEM,"+
                "handover_code                                                    AS ASSOCIATE_NURSE_ID,"+
                "handover_name                                                    AS ASSOCIATE_NURSE_SIGN_NAME,"+
                "handover_signdate                                                AS ASSOCIATE_NURSE_SIGN_DATETIME,"+
                "s.shipper_signdate                                                 AS TRANSPORT_STAFF_SIGN_DATETIME,"+
                "s.shipper_code                                                     AS TRANSPORT_STAFF_ID,"+
                "s.shipper_name                                                     AS TRANSPORT_STAFF_SIGN_NAME,"+
                "visit_date                                                       AS VISIT_DATE,"+
                "bed_no                                                           AS BED_NO,"+
                "ward_id                                                          AS ROOM_NO,"+
                "dept_code                                                        AS DEPT_ID,"+
                "dept_name                                                        AS DEPT_NAME,"+
                "wards_id                                                         AS WARD_ID,"+
                "wards_name                                                       AS WARD_NAME,"+
                "weight                                                           AS PATIENT_WEIGHT,"+
                "abo_code                                                         AS ABO_BLOOD_CODE,"+
                "abo                                                              AS ABO_BLOOD_NAME,"+
                "rh_code                                                          AS RH_BLOOD_CODE,"+
                "rh                                                               AS RH_BLOOD_NAME,"+
                "nursing_gradecode                                                AS NURSE_GRADE_CODE,"+
                "nursing_typecode                                                 AS NURSE_TYPE_CODE,"+
                "nursing_type                                                     AS NURSE_TYPE_NAME,"+
                "patrol_date_ao                                                   AS PATROL_DATE_AO,"+
                "patrol_code_ao                                                   AS PATROL_CODE_AO,"+
                "patrol_name_ao                                                   AS PATROL_NAME_AO,"+
                "scrub_date_ao                                                    AS SCRUB_DATE_AO,"+
                "scrub_code_ao                                                    AS SCRUB_CODE_AO,"+
                "scrub_name_ao                                                    AS SCRUB_NAME_AO,"+
                "oper_good_ao                                                     AS OPER_GOOD_AO,"+
                "preoperative_counting_mark                                       AS PRE_OPER_STOCKTAKE_FLAG,"+
                "patrol_date_ac                                                   AS PATROL_DATE_AC,"+
                "patrol_code_ac                                                   AS PATROL_CODE_AC,"+
                "scrub_code_ac                                                    AS SCRUB_CODE_AC,"+
                "scrub_name_ac                                                    AS SCRUB_NAME_AC,"+
                "oper_good_ac                                                     AS IN_OPER_GOODS_NAME,"+
                "before_check_mark                                                AS BEFORE_CHECK_MARK,"+
                "patrol_date_bc                                                   AS PATROL_DATE_BC,"+
                "patrol_code_bc                                                   AS PATROL_CODE_BC,"+
                "patrol_name_bc                                                   AS PATROL_NAME_BC,"+
                "scrub_code_bc                                                    AS SCRUB_CODE_BC,"+
                "scrub_name_bc                                                    AS SCRUB_NAME_BC,"+
                "oper_good_bc                                                     AS OPER_GOOD_BC,"+
                "after_check_mark                                                 AS AFTER_CHECK_MARK,"+
                "oper_start_date                                                  AS OPER_START_DATETIME,"+
                "oper_end_date                                                    AS OPER_END_DATETIME,"+
                "surgeon_code                                                     AS OPER_DOCTOR_ID,"+
                "surgeon_name                                                     AS OPER_DOCTOR_NAME,"+
                "target_code                                                      AS OPER_SITE_CODE,"+
                "target_name                                                      AS OPER_SITE_NAME,"+
                "oper_code                                                        AS OPER_CODE,"+
                "oper_name                                                        AS OPER_NAME,"+
                "admit_room_date                                                  AS ADMIT_ROOM_DATE,"+
                "out_oper_date                                                    AS OUT_OPER_DATE,"+
                "preparations                                                     AS PREPARATIONS," +
                "visit_id                                                         AS VISIT_ID "+
                "from hdsd00_08_05 s " +
                "left join tdy_list_oper b on b.unique_id=s.unique_id and b.xds_type=s.xds_type " +
                "left join tdy_list_handover c on c.unique_id=s.unique_id and c.xds_type=s.xds_type");
        tableEnv.createTemporaryView("join_result",joinResult);
    }

    private void createSourceTables(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table hdsd00_08_05 (" +
                "pk                             String,"+
                "upload_time                    String,"+
                "status                         String,"+
                "empi                           String,"+
                "encounter_id                   String,"+
                "bed_name                       String,"+
                "ward_name                      String,"+
                "patient_id                     String,"+
                "patient_domain                 String,"+
                "visit_domain                   String,"+
                "visit_id                       String,"+
                "visit_times                    String,"+
                "unique_id                      String,"+
                "xds_type                       String,"+
                "xds_name                       String,"+
                "effective_time                 String,"+
                "xds_version                    String,"+
                "patient_type                   String,"+
                "patient_typecode               String,"+
                "out_patient_id                 String,"+
                "domain_id                      String,"+
                "inpatient                      String,"+
                "identify_no                    String,"+
                "patient_name                   String,"+
                "sex                            String,"+
                "sex_code                       String,"+
                "age                            String,"+
                "age_unit                       String,"+
                "institutional_incode           String,"+
                "institutional_in               String,"+
                "write_time                     String,"+
                "author_code                    String,"+
                "author_name                    String,"+
                "custodian_code                 String,"+
                "custodian_name                 String,"+
                "patrol_signdate                String,"+
                "patrol_code                    String,"+
                "patrol_name                    String,"+
                "instrument_signdate            String,"+
                "instrument_code                String,"+
                "instrument_name                String,"+
                "handover_code                  String,"+
                "handover_name                  String,"+
                "handover_signdate              String,"+
                "shipper_signdate               String,"+
                "shipper_code                   String,"+
                "shipper_name                   String,"+
                "visit_date                     String,"+
                "bed_no                         String,"+
                "ward_id                        String,"+
                "dept_code                      String,"+
                "dept_name                      String,"+
                "wards_id                       String,"+
                "wards_name                     String,"+
                "hospital_code                  String,"+
                "hospital_name                  String,"+
                "weight                         String,"+
                "abo_code                       String,"+
                "abo                            String,"+
                "rh_code                        String,"+
                "rh                             String,"+
                "skin_check                     String,"+
                "allergy_history_mark           String,"+
                "allergy_history                String,"+
                "nursing_gradecode              String,"+
                "nursing_grade                  String,"+
                "nursing_typecode               String,"+
                "nursing_type                   String,"+
                "patrol_date_ao                 String,"+
                "patrol_code_ao                 String,"+
                "patrol_name_ao                 String,"+
                "scrub_date_ao                  String,"+
                "scrub_code_ao                  String,"+
                "scrub_name_ao                  String,"+
                "oper_good_ao                   String,"+
                "preoperative_counting_mark     String,"+
                "patrol_date_ac                 String,"+
                "patrol_code_ac                 String,"+
                "patrol_name_ac                 String,"+
                "scrub_date_ac                  String,"+
                "scrub_code_ac                  String,"+
                "scrub_name_ac                  String,"+
                "oper_good_ac                   String,"+
                "before_check_mark              String,"+
                "patrol_date_bc                 String,"+
                "patrol_code_bc                 String,"+
                "patrol_name_bc                 String,"+
                "scrub_date_bc                  String,"+
                "scrub_code_bc                  String,"+
                "scrub_name_bc                  String,"+
                "oper_good_bc                   String,"+
                "after_check_mark               String"+
                ")" +
                FlinkSourceUtil.getKafkaWith("hdsd00_08_05","App09_DwdMhsSurgicalNursingRecord"));

        super.readTdyListHandover(tableEnv,"App09_DwdMhsSurgicalNursingRecord");
        super.readTdyListOper(tableEnv,"App09_DwdMhsSurgicalNursingRecord");
    }
}
